package com.jokerku.mini.schedule.leader;

import com.jokerku.mini.schedule.common.Constants;
import com.jokerku.mini.schedule.domain.ZkPath;
import com.jokerku.mini.schedule.event.LeaderEventListener;
import com.jokerku.mini.schedule.service.ZkCuratorServer;
import com.jokerku.mini.schedule.util.SleepUtil;
import org.apache.curator.framework.CuratorFramework;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @Author: guzq
 * @CreateTime: 2023/06/01 23:13
 * @Description: leader服务
 * @Version: 1.0
 */
public class LeaderService {

    private static final Logger logger = LoggerFactory.getLogger(LeaderService.class);

    private final CuratorFramework client;
    private final String path;

    public LeaderService(CuratorFramework client, String path) {
        this.client = client;
        this.path = path;
    }

    public void selectLeader() throws Exception {
        LeaderShipSelector shipSelector = new LeaderShipSelector(client, path, new LeaderExecExecutionCallback());
        shipSelector.select();
    }

    public boolean isLeader() {
        boolean hasLeader = hasLeader();
        while (!hasLeader) {
            SleepUtil.sleep();
            hasLeader = hasLeader();
        }
        try {
            if (Constants.Global.ip.equals(ZkCuratorServer.getData(client, ZkPath.getMasterInstancePath()))) {
                return true;
            }
        } catch (Exception e) {
            logger.error("Check is leader error", e);
        }
        return false;
    }

    public boolean hasLeader() {
        return ZkCuratorServer.isExisted(client, ZkPath.getMasterInstancePath());
    }

    class LeaderExecExecutionCallback implements LeaderExecutionCallback {
        @Override
        public void callback() throws Exception {
            if (!hasLeader()) {
                logger.info("当前 {} 成为主节点", Constants.Global.ip);
                ZkCuratorServer.createEphemeralNode(client, ZkPath.getMasterInstancePath(), Constants.Global.ip);
            }
            // 监听主节点下线事件
            ZkCuratorServer.addTreeCacheListener(client, ZkPath.getMasterInstancePath(), new LeaderEventListener());
        }
    }
}
